package com.rrd.dw.mr.gzip.kafka;

import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.tianxia.TxCreditReportAdditionalRecord;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;


/*
 * 天下信用报告记录-主表-新增字段
 *
 * @author:szw
 * */

public class TxCreditMainAdditionalParse extends ParseBase {
    private Text key = new Text();
    private Text val = new Text();

    /**
     * 天下信用报告记录-主表-新增字段
     *
     * @param context
     * @param model
     * @param successful
     * @return boolean
     * @throws IOException
     * @throws InterruptedException
     * @author typole
     * @date 2020/11/30 5:08 下午
     */

    @Override
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        TxCreditReportAdditionalRecord objAdditional = (TxCreditReportAdditionalRecord) model;

        StringBuilder sb = getBaseInfo(objAdditional, successful);
        StringBuilder TxSb = new StringBuilder(2000);

        // isTriggerItem 是否触发提额
        TxSb.append(objAdditional.getIsTriggerItem() == null ? "" : objAdditional.getIsTriggerItem()).append(CRT001);
        // isDoubtful 是否有前端校验失败记录
        TxSb.append(objAdditional.getIsDoubtful() == null ? "" : objAdditional.getIsDoubtful()).append(CRT001);
        // uploadReportType 报告来源区分标签
        TxSb.append(objAdditional.getUploadReportType() == null ? "" : objAdditional.getUploadReportType()).append(CRT001);
        // type 提额类型
        TxSb.append(objAdditional.getType() == null ? "" : objAdditional.getType()).append(CRT001);
        // userKey 用户 key
        TxSb.append(objAdditional.getUserKey() == null ? "" : objAdditional.getUserKey()).append(CRT001);
        key.set(getKey());
        val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
        context.write(key, val);
        return true;
    }

    @Override
    public String getSubType() {
        return ".main_additional";
    }

    @Override
    public String getTableName() {
        return "kafka_babel_tx_credit_main_additional_inc";
    }

    @Override
    public String getType() {
        return RecordType.TX_CREDIT_REPORT_ADDITIONAL_RECORD.toString();
    }

    @Override
    public String getComment() {
        return "天下信用-主表-新增字段";
    }
}

